package com.atguigu.sparkstreaming.demos

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/7/15
 *
 *    无状态的运算:  每个批次之间是相互独立的，各算各的。
 *    有状态的运算:   当前批次的计算需要使用到上一个批次运算的结果，在上一次运算的基础上继续运算。
 *                      无需掌握。
 *                      SparkStreaming原生的计算方式有很大的弊端，虽然提供了有状态计算的算子，但是没有人用。
 *                        DStream.xxxStatexxx(): SparkStreaming提供的有状态的计算的算子。
 *
 *                        后续使用其他的方式解决这个问题。
 *
 *     ------------------------
 *     select
 *        a,  xxx(b)   ----UDF(输入1行，输出1行)  map
 *
 *     from xxx
 *     where xxxx     --------filter
 *     group by       -------- redueceByKey | groupByKey
 *     order by       -------- sortBy | sortByKey
 *     limit x         -------- take(x)
 *
 *    -----------------------
 *    transform:
 *        def transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U]
 *              将 DStream[T] 中的RDD[T]取出来  调用transformFunc ，转换为 RDD[U]，封装为DStream[U]
 *
 *         算子:     RDD.map，RDD.filter
 *         抽象原语： DStream.map  DStream.filter
 *
 *         算子远比抽象原语丰富，有些方法算子有，但是抽象原语没有。如果希望使用此类方法，只能
 *         将对DStream的运算转换为对其中RDD的运算，才能调用 RDD的算子，算完后，再把结果封装为DStream
 *
 *
 */
object TransformDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "TransformDemo", Seconds(10))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "220309",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //(String, Int):(word,n)
    val ds1: DStream[(String, Int)] = ds.flatMap(record => record.value().split(" "))
      .map((_, 1)).reduceByKey(_+_)

    //对单次进行sortByKey操作
    val ds2: DStream[(String, Int)] = ds1.transform(rdd => rdd.sortByKey())

    ds2.print(1000)

    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
